//package com.kingsoft.dc.khaos.module.spark.sink
//
//import com.kingsoft.dc.khaos.KhaosContext
//import com.kingsoft.dc.khaos.innertype.Schema
//import com.kingsoft.dc.khaos.module.spark.constants.HDFSEnum
//import com.kingsoft.dc.khaos.module.spark.metadata.sink.HdfsConfig
//import com.kingsoft.dc.khaos.util.Logging
//import org.apache.spark.sql.functions.concat_ws
//import org.apache.spark.sql.types.StringType
//import org.apache.spark.sql.{Column, DataFrame, SaveMode}
//import org.json4s.{DefaultFormats, JsonAST}
//
//import scala.collection.mutable.ArrayBuffer
//
///**
//  * create by yansu on 2019/09/11 16:18
// * 不修改设置默认值逻辑
//  */
//class DFSSink extends SinkStrategy with Logging {
//  /** 数据输出 */
//  override def sink(kc: KhaosContext,
//                    module_id: String,
//                    config: JsonAST.JObject,
//                    schema: Schema,
//                    dataSet: DataFrame): Any = {
//
//    implicit val formats = DefaultFormats
//    val hdfsInfo = config.extract[HdfsConfig]
//    //var resDF = DataframeUtils.setDefaultValue(hdfsInfo.extract_fields,null,dataSet)
//
//    val colArr = new ArrayBuffer[Column]()
//    for (elem <- hdfsInfo.map_fields) {
//      colArr += dataSet.col(elem.field)
//    }
//    val value: DataFrame = dataSet.select(colArr: _*)
//
//    val sinkPathStr = hdfsInfo.path
//    val dataType = hdfsInfo.pattern
//
//    if (sinkPathStr == "" || sinkPathStr == null) {
//      throw new IllegalArgumentException(s"HDFS路径不能为空!")
//    }
//
//    if (dataType != null && dataType != "") {
//      dataType.trim.toLowerCase match {
//        case HDFSEnum.ORC => {
//          value.write.mode(SaveMode.Append).orc(sinkPathStr)
//        }
//        case HDFSEnum.TEXT => {
//          value.select(concat_ws(hdfsInfo.delimit, colArr: _*).cast(StringType)).write.mode(SaveMode.Append).text(sinkPathStr)
//        }
//        case HDFSEnum.PARQUET => {
//          value.write.mode(SaveMode.Append).parquet(sinkPathStr)
//        }
//        case _ => {
//          throw new IllegalArgumentException(s"暂不支持存储 ${dataType} 类型数据!")
//        }
//      }
//    }
//
//  }
//}
